package SQL

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StructField, StructType}
import org.junit.Test

class 缺失值的处理 {
  val spatk = SparkSession.builder()
    .master("local[6]")
    .appName("null processor")
    .getOrCreate()

  import spatk.implicits._
  import org.apache.spark.sql.functions._

  @Test
  //缺失数字类型
  def NullandNan(): Unit = {
    val schema = StructType(
      Seq(
        StructField("id", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("season", IntegerType),
        StructField("pm", DoubleType)
      )
    )
    val source = spatk.read
      .schema(schema)
      .option("header", value = true)
      .csv("E:\\data\\spark数据\\Spark_data\\Beijing.csv")
    //只要有一个为空，那就删除一行的内容
    source.na.drop("any").show()
    //只有在全部为空的时候，才删除
    source.na.drop("all").show()
    //自定义的一些规则,在集合里的数据任何一个是空的，就删除
    source.na.drop("any", List("year", ",month", "day", "hour")).show()

    //缺失值得填充
    //针对于所有的数据进行填充
    source.na.fill(0).show()
    //针对于特定的列
    source.na.fill(0, List("year", "month"))
  }

  @Test
  def strProcessor(): Unit = {
    val schema = StructType(
      Seq(
        StructField("id", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("season", IntegerType),
        StructField("pm", DoubleType)
      )
    )
    val source = spatk.read
      .schema(schema)
      .option("header", value = true)
      .csv("E:\\data\\spark数据\\Spark_data\\Beijing.csv")

    //丢弃
    source.where('pm =!= "NaN").show()
    //替换
    //如果pm大于NaN，那么就把他转换成Double.NaN
    //如果不是NaN，那么就把pm转换成double类型
    source.select(
      'id,
      when('pm === "NaN", Double.NaN)
        .otherwise('pm cast (DoubleType))
        .as("pm")
    ).show()
    //replace    原类型和转换的类型必须一致
    //将pm中所有NaN替换成NA  NULL转成小写
    source.na.replace("pm", Map("NaN" -> "NA","NULL"->"null"))
  }
}
